package com.we.risk.antirefund.refund;

import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.*;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Iterator;

public class TagOrderInfo {
    public long uid;
    public int refund_times;

    public TagOrderInfo() {}

    public TagOrderInfo(long uid, int refund_times) {
        this.uid = uid;
        this.refund_times = refund_times;
    }

    public static TagOrderInfo init(long uid) {
        return new TagOrderInfo(uid, 0);
    }

    public static Logger LOG = LoggerFactory.getLogger(TagOrderInfo.class);

    public static SingleOutputStreamOperator<String> getFlatMapStream(
            StreamTableEnvironment tableEnv, StreamExecutionEnvironment env) {
        Table tableRes = tableEnv.sqlQuery(TableOrderInfo.QUERYSQL);
        DataStream<Tuple2<Boolean, TableOrderInfo>> inputStream =
                tableEnv.toRetractStream(tableRes, TableOrderInfo.class);

        SingleOutputStreamOperator<String> out = inputStream
                .filter(new FilterFunction<Tuple2<Boolean, TableOrderInfo>>() {
                    @Override
                    public boolean filter(Tuple2<Boolean, TableOrderInfo> value) throws Exception {
                        return value.f0;
                    }
                })
                .uid("TagOrderInfo-filter")
                .keyBy(new KeySelector<Tuple2<Boolean, TableOrderInfo>, Long>() {
                    @Override
                    public Long getKey(Tuple2<Boolean, TableOrderInfo> value) throws Exception {
                        return value.f1.getUid();
                    }
                })
                .flatMap(new RichFlatMapFunction<Tuple2<Boolean, TableOrderInfo>, String>() {
                    ValueState<TagOrderInfo> orderNoVState;
                    MapState<String, Integer> orderMState;
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        orderNoVState = getRuntimeContext()
                                .getState(new ValueStateDescriptor<TagOrderInfo>("orderNoVState", TagOrderInfo.class));
                        orderMState = getRuntimeContext()
                                .getMapState(new MapStateDescriptor<String, Integer>("orderMState", String.class, Integer.class));
                    }

                    @Override
                    public void flatMap(Tuple2<Boolean, TableOrderInfo> value, Collector<String> out) throws Exception {
                        long uid = value.f1.getUid();
                        String order_no = value.f1.getOrder_no();

                        TagOrderInfo tmpState = orderNoVState.value();
                        TagOrderInfo curState = (tmpState == null ? TagOrderInfo.init(uid) : tmpState);

                        //Mstate 仅用来去重,通过单独的变量进行计数，避免遍历计数
                        if(!orderMState.contains(order_no)) {
                            orderMState.put(order_no, 1);

                            int curTimes = curState.getRefund_times() + 1;
                            JSONObject outJson = new JSONObject();
                            outJson.put("uid", uid);
                            outJson.put("refund_times", curTimes);
                            out.collect(outJson.toString());

                            curState.setRefund_times(curTimes);
                            orderNoVState.update(curState);
                        }
                    }
                });
        return out;
    }

    public long getUid() {
        return uid;
    }

    public int getRefund_times() {
        return refund_times;
    }

    public void setRefund_times(int refund_times) {
        this.refund_times = refund_times;
    }
}
